package com.kasinf.sample.kafka.consumer;

import com.kasinf.sample.kafka.message.TestMessages;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * 消息消费者
 *
 * @Author : Kaiser
 * @Date: 2025/5/28  16:14
 */
@Slf4j
@Component
public class MessageConsumer {

    @KafkaListener(topics = "KS_TEST_TOPIC", groupId = "KS_TEST_TOPIC_CONSUMER")
    @Transactional
    public void onMessage(TestMessages message, Acknowledgment acknowledgment) {
        try {
            log.info("[onMessage][消息队列消息({})]", message.getMessage());
            // ... 处理具体逻辑

            // 手动提交
            acknowledgment.acknowledge();
        } catch (Exception e) {
            // 事务一定要抛出异常才会回归
            throw new RuntimeException(e.getMessage());
        }
    }


    @KafkaListener(topics = "KS_BOARD_TOPIC", containerFactory = "boardListenerContainer")
    public void onMessage2(TestMessages message) {
        log.info("[onMessage2][广播消息({})]", message.getMessage());
    }

}
